package com.seari.movedata.main.dataHander;

import com.seari.movedata.service.FirstInitServiceImpl;
import com.seari.movedata.utils.CommonUtils;
import com.seari.movedata.utils.JedisPoolUtils;
import com.seari.movedata.utils.ListUtils;
import com.seari.movedata.utils.PropertiesUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

@Component
@Order(value = 2)
public class HistoryDataHandlerStarter implements ApplicationRunner {

    final static Logger logger = LoggerFactory.getLogger(HistoryDataHandlerStarter.class);

    public static final int MAX_WORKER_COUNT = 10;
    public static final int MAX_SLICE_COUNT = 20000;
    public static BlockingQueue<List<String>> aiQueue = new LinkedBlockingQueue<>(200);
    public static BlockingQueue<List<String>> diQueue = new LinkedBlockingQueue<>(200);
    public static ExecutorService service = Executors.newCachedThreadPool();
    public static AtomicInteger workerCount = new AtomicInteger();
    Jedis jedis = JedisPoolUtils.getJedis();

    List<String> aiKeyList = new ArrayList<String>();
    List<String> diKeyList = new ArrayList<String>();

    FirstInitServiceImpl firstInitServiceImpl = new FirstInitServiceImpl();

    @Override
    public void run(ApplicationArguments args) throws Exception{

        initKeyList();
        logger.info("init key complete");
        logger.info("aiKeyList=  "+aiKeyList);
        logger.info("diKeyList=  "+diKeyList);

        initLocalCacheAndPutDataToInfluxdb(aiKeyList,diKeyList);
        TimeUnit.SECONDS.sleep(60);
        logger.info("first init localCache and putData To influxdb complete");

        while (true) {
            for(int i=0;i<aiKeyList.size();i++) {
                String aikeyStr = aiKeyList.get(i);
                Set<String> aiKeyset = jedis.hkeys(aikeyStr);
                if(!CommonUtils.objectIsNull(aiKeyset)) {
                    List<String> result = new ArrayList<String>(aiKeyset);
                    List<List<String>> listList = ListUtils.splitList(result, MAX_SLICE_COUNT);
                    for (List<String> list : listList) {
                        List<String> listStr = new ArrayList<String>();
                        listStr.addAll(list);
                        listStr.add(aikeyStr);
                        aiQueue.add(listStr);
                    }
                }else{
                    logger.info( "   "+aikeyStr+ " key  not find in redis");
                }
            }

            for(int i=0;i<diKeyList.size();i++) {
                String dikeyStr = diKeyList.get(i);
                Set<String> diKeyset = jedis.hkeys(dikeyStr);
                if(!CommonUtils.objectIsNull(diKeyset)) {
                    List<String> result = new ArrayList<String>(diKeyset);
                    List<List<String>> listList = ListUtils.splitList(result, MAX_SLICE_COUNT);
                    for (List<String> list : listList) {
                        List<String> listStr = new ArrayList<String>();
                        listStr.addAll(list);
                        listStr.add(dikeyStr);
                        diQueue.add(listStr);
                    }
                }else{
                   logger.info( " "+dikeyStr+ "  key not find in redis");
                }
            }
            startWorker();
            TimeUnit.SECONDS.sleep(60);
        }
    }

    public void startWorker(){

        while(workerCount.get() < MAX_WORKER_COUNT){
            HistoryDataHandlerWorker worker = new HistoryDataHandlerWorker(aiQueue,diQueue);
            service.submit(worker);
            int num = workerCount.incrementAndGet();
            logger.info("start worker:current work count is:" + num);
        }
    }

    public  void initKeyList(){

        List<String> keyAiList = new ArrayList<>();
        String keyAi = PropertiesUtil.getProperty("commonConfig.properties","keyAi");
        aiKeyList = Arrays.asList(keyAi.split(","));

        List<String> keyDiList = new ArrayList<>();
        String keyDi = PropertiesUtil.getProperty("commonConfig.properties","keyDi");
        diKeyList = Arrays.asList(keyDi.split(","));
    }

    public void initLocalCacheAndPutDataToInfluxdb(List<String> aiKeyList,List<String> diKeyList){

        firstInitServiceImpl.initLocalCacheAndPutDataToInfluxdb(aiKeyList,diKeyList);
    }
}
